/*
 * =====================================================================================
 *
 *       Filename:  worker_queue.c
 *
 *    Description:  
 *
 *        Version:  1.0
 *        Created:  05/21/2011 03:18:17 PM
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  Ken Chen (Ken), chen_kenz@emc.com
 *        Company:  EMC
 *
 * =====================================================================================
 */

#include <assert.h>
#include <stdio.h>
#include "worker_queue.h"

static int num_of_pending_jobs = 0;
/* Not thread safe, assume the lock is already held */
static void inline do_delete_worker(struct worker_entry* went)
{
  struct worker_entry_list* deleted_entry = (struct worker_entry_list*)went;
  deleted_entry->prev->next = deleted_entry->next;
  deleted_entry->next->prev = deleted_entry->prev;
  deleted_entry->prev = NULL;
  deleted_entry->next = NULL;
}

void init_worker_queue(struct worker_queue* wq) 
{
  assert(wq != NULL);
  pthread_mutex_init(&wq->worker_lock, NULL);
  pthread_cond_init(&wq->worker_cond, NULL);
  wq->pending_jobs.prev = wq->pending_jobs.next = &wq->pending_jobs;
}

void init_job(struct worker_entry* went, 
              worker_callback func, 
              void* data) 
{
  assert(went != NULL);
  went->pending_jobs.prev = went->pending_jobs.next = NULL;
  went->callback = func;
  went->data = data;
}

int add_job(struct worker_queue* wq, struct worker_entry* went) 
{
  struct worker_entry_list* new_job = (struct worker_entry_list*)went;
  assert((wq != NULL) && (new_job != NULL)); 
  pthread_mutex_lock(&wq->worker_lock);
  /* Add the job to the tail of the pending job list */
  wq->pending_jobs.prev->next = new_job;
  new_job->prev = wq->pending_jobs.prev;  
  new_job->next = &wq->pending_jobs;
  wq->pending_jobs.prev = new_job;
  /* Notify do_jobs, we have a new job to do */
  ++num_of_pending_jobs;
  pthread_cond_signal(&wq->worker_cond);
  pthread_mutex_unlock(&wq->worker_lock);
}

int do_jobs(struct worker_queue* wq) 
{
  struct worker_entry* went;
  worker_callback work_func;
  void* work_data;
  assert(wq != NULL);
  for(;;)
  {
    pthread_mutex_lock(&wq->worker_lock);
    while(wq->pending_jobs.next == &wq->pending_jobs)
    {
      pthread_cond_wait(&wq->worker_cond, &wq->worker_lock);   
    }
    went = (struct worker_entry*)(wq->pending_jobs.next);
    work_func = went->callback; 
    work_data = went->data;
    do_delete_worker(went); 
    --num_of_pending_jobs;
    pthread_mutex_unlock(&wq->worker_lock);
    work_func(work_data);
    printf("pending jobs: %d\n", num_of_pending_jobs);
  }
}

